package testing.ms2;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import junit.framework.TestCase;

import org.junit.Test;

import client.KVStore;
import common.messages.KVMessage;
import common.messages.KVMessageImpl;

public class AdditionalTest extends TestCase {
	
	// TODO add your test cases, at least 3
		
	public void setUp() {
	}

	public void tearDown() {
	}
	
	
	@Test
	public void testPutStressTest() {
		System.out.println("TEST: PUT stress test running ... ");
		String address = "localhost";
		int port = 50000;
		int numberOfClients = 40;
		Exception ex = null;
		List<KVStore> clients = new ArrayList<KVStore>();
		for(int i = 0; i<numberOfClients; i++){
			KVStore client = new KVStore(address, port,false);
			clients.add(client);
		}
		
		ExecutorService threadPool = Executors.newFixedThreadPool(5 + (numberOfClients / 10));
		CompletionService<String> pool = new ExecutorCompletionService<String>(threadPool);
		int index = 0;
		for(final KVStore client : clients){
			final int keyIndex = index ;
			pool.submit(new Callable<String>() {
				public String call() {
					String result = "";
					String kk = "key"+keyIndex;
					try {
						client.connect();
						KVMessage response = client.put(kk, kk);
						result = response.getStatus().name();
						client.disconnect();
					} catch (Exception e) {
						result = KVMessage.StatusType.PUT_ERROR.name();
					}
					return result;
				}
			});
			index ++ ;
		}

		boolean finalResult = true;
		for(int i = 0; i < numberOfClients; i++){
			String result = "";
			try {
				result = pool.take().get();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
			//System.out.println(">"+result);
			if(!result.equals(KVMessage.StatusType.PUT_SUCCESS.name())){
				finalResult = false;
			}
		   //Compute the result
		}

		threadPool.shutdown();
		System.out.println("TEST: PUT stress test finished! - " + finalResult);
		assertTrue(finalResult);		
	}
	
	@Test
	public void testGetStressTest() {
		System.out.println("TEST: PUT stress test running ... ");
		String address = "localhost";
		int port = 50000;
		int numberOfClients = 40;
		Exception ex = null;
		KVStore masterClient = new KVStore(address, port,false);
		try {
			masterClient.connect();
			masterClient.put("kk", "vv");
		} catch (Exception e1) {
			e1.printStackTrace();
			assertFalse(false);
			return;
		}
		
		List<KVStore> clients = new ArrayList<KVStore>();
		for(int i = 0; i<numberOfClients; i++){
			KVStore client = new KVStore(address, port,false);
			clients.add(client);
		}
		
		ExecutorService threadPool = Executors.newFixedThreadPool(5 + (numberOfClients / 10));
		CompletionService<String> pool = new ExecutorCompletionService<String>(threadPool);
		int index = 0;
		for(final KVStore client : clients){
			final int keyIndex = index ;
			pool.submit(new Callable<String>() {
				public String call() {
					String result = "";
					String kk = "kk";
					try {
						client.connect();
						client.put("k"+keyIndex, "v1" + keyIndex);
						KVMessage response = client.get(kk);
						result = response.getValue();
						client.disconnect();
					} catch (Exception e) {
						result = KVMessage.StatusType.GET_ERROR.name();
					}
					return result;
				}
			});
			index ++ ;
		}

		boolean finalResult = true;
		for(int i = 0; i < numberOfClients; i++){
			String result = "";
			try {
				result = pool.take().get();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
			//System.out.println(">"+result);
			if(!result.equals("vv")){
				finalResult = false;
			}
		   //Compute the result
		}

		threadPool.shutdown();
		System.out.println("TEST: GET stress test finished! - " + finalResult);
		assertTrue(finalResult);		
	}
	

	@Test
	public void testUpdateStressTest() {
		System.out.println("TEST: PUT stress test running ... ");
		String address = "localhost";
		int port = 50000;
		int numberOfClients = 10;
		Exception ex = null;
		List<KVStore> clients = new ArrayList<KVStore>();
		for(int i = 0; i<numberOfClients; i++){
			KVStore client = new KVStore(address, port, false);
			clients.add(client);
		}
		/** Propagate the server with a test value **/
		final String stressKey = "testUpdateStressTest";
		KVStore putValueClient = new KVStore(address, port, false);
		try {
			putValueClient.connect();
			putValueClient.put(stressKey, "initial value");
			putValueClient.disconnect();
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		
		/** Send update commands from multiple clients at once **/
		ExecutorService threadPool = Executors.newFixedThreadPool(5 + (numberOfClients / 10));
		CompletionService<String> pool = new ExecutorCompletionService<String>(threadPool);
		int index = 0;
		for(final KVStore client : clients){
			final int keyIndex = index ;
			pool.submit(new Callable<String>() {
				public String call() {
					String result = "";
					try {
						client.connect();
						KVMessage response = client.put(stressKey, "Update from "+keyIndex);
						result = response.getStatus().name();
						client.disconnect();
					} catch (Exception e) {
						result = KVMessage.StatusType.PUT_ERROR.name();
					}
					return result;
				}
			});
			index ++ ;
		}

		boolean finalResult = true;
		for(int i = 0; i < numberOfClients; i++){
			String result = "";
			try {
				result = pool.take().get();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
			//System.out.println(">"+result);
			if(!result.equals(KVMessage.StatusType.PUT_UPDATE.name())){
				finalResult = false;
			}
		   //Compute the result
		}

		threadPool.shutdown();
		System.out.println("TEST: UPDATE stress test finished! - " + finalResult);
		assertTrue(finalResult);		
	}
}
